package sg.edu.nus.iss.stockex;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class SectorPerformanceMapReduceJob {
	public static HashMap<String,String> sectorMap= null;
	public static class ProfitMap extends Mapper<LongWritable, Text, Text, Stock> {
		
//	private final static Company one = new Company();
	private Text word = new Text();

	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		try{
			String line = value.toString();
			System.out.println("Line is: "+line);
			Stock stock = Stock.parseString(line);
			word.set(stock.getCode());
			context.write(word, stock);
		}
		catch(Exception ex)
		{
			System.out.println(ex.getMessage());
			ex.printStackTrace();
		}
	}
}
	public static class ProfitReduce extends
	Reducer<Text, Stock, Text, LongWritable> {
		
		public void reduce(Text key, Iterable<Stock> values,
			Context context) throws IOException, InterruptedException {
		double profit = 0;
		double open = 0;
		double close = 0;
		long volume = 0;
		Text stockText = new Text();
		
		for(Stock stock : values)
		{
			open = stock.getOpenPrice();
			close = stock.getLowPrice();
			volume = stock.getVolume();
			profit += (open - close) * volume;
			System.out.println("Open: "+open + " Close: "+ close+" volume: "+ volume + " profit: "+ profit);

		}
		System.out.println("profit for "+key + " quote is: " + profit);
		LongWritable prof = new LongWritable((long)profit);
		context.write(key, prof);
		}
	}
	
	public static class SectorPerformanceMap extends Mapper<LongWritable, Text, Text, LongWritable> {
		
//		private final static Company one = new Company();
		private Text word = new Text();

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			try{
				String line = value.toString();
				String[] data = line.split(" ");
				if(data.length==2){
				String code = data[0];
				String prof = data[1];
				long profit = new Long(prof.trim()).longValue();
				
				System.out.println("profit for code "+code +"is : "+profit);
				if(sectorMap.containsKey(code))
				{
					word.set(sectorMap.get(code));
					context.write(word, new LongWritable(profit));
				}
				
				}
			}
			catch(Exception ex)
			{
				System.out.println("lll"+ex.getMessage());
				ex.printStackTrace();
			}
		}
	}
	
	public static class SectorPerformanceReduce extends
	Reducer<Text, LongWritable, Text, LongWritable> {
		
		public void reduce(Text key, Iterable<LongWritable> values,
			Context context) throws IOException, InterruptedException {
		
		long profit = 0;
		
		Text stockText = new Text();
	
		
		for(LongWritable longWritable  : values)
		{
			profit += longWritable.get();
		}
		
		System.out.println("total net profit for sector "+ key.toString() +"is:" + profit);
		context.write(key, new LongWritable(profit));
		}
	}
	public static void startJob(int quarter, int year) throws Exception
	{
		Configuration conf = new Configuration();
		
		conf.setInt(CONFIG.QUARTER, quarter);
		conf.setInt(CONFIG.YEAR,year);
		Job job = new Job(conf, "StockEx");
		// output of the Map class
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Stock.class);
		// output of the job..
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);

		job.setMapperClass(ProfitMap.class);
		job.setReducerClass(ProfitReduce.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.addInputPath(job, new Path("input"));
		// Erase previous run output (if any)
		FileSystem.get(conf).delete(new Path("temp"), true);
		FileOutputFormat.setOutputPath(job, new Path("temp"));
		try{
		boolean status =  job.waitForCompletion(true);
		if(status)
		{
			Job job1 = new Job(conf, "StockEx1");
			// output of the Map class
			job1.setMapOutputKeyClass(Text.class);
			job1.setMapOutputValueClass(LongWritable.class);
			// output of the job..
			job1.setOutputKeyClass(Text.class);
			job1.setOutputValueClass(LongWritable.class);

			job1.setMapperClass(SectorPerformanceMap.class);
			job1.setReducerClass(SectorPerformanceReduce.class);

			job1.setInputFormatClass(TextInputFormat.class);
			job1.setOutputFormatClass(TextOutputFormat.class);

			FileInputFormat.addInputPath(job1, new Path("temp"));
			// Erase previous run output (if any)
			FileSystem.get(conf).delete(new Path("output"), true);
			FileOutputFormat.setOutputPath(job1, new Path("output"));
			boolean finalStatus =  job1.waitForCompletion(true);
			System.out.println("final status of the job is: " + finalStatus);
		}
		}catch(Exception e)
		{
			System.out.println("job"+ e.getMessage());
			e.printStackTrace();
		}
			}
	public static void main(String[] args) throws Exception {
		File file = new File("/home/hadoop/workspace/sector-mapping.txt");
		FileReader reader = new FileReader(file);
		BufferedReader br = new BufferedReader(reader);
		String data = null;
		String code = "";
		String sector ="";
		String[] rec;
		sectorMap = new HashMap<String,String>();
		while((data=br.readLine())!="" &&  data!=null)
		{
			System.out.println(data);
			 rec = data.split(";");
			 if(rec.length!=2) return;
			 code = rec[0];
			 sector = rec[1];
			 if(!sectorMap.containsKey(sector))
			 {
				 if(!(code.equals("") || code==null))
				 {
					 sectorMap.put(code, sector);
				 }
			 }
		}
		int quarter = 2;
		int year = 2012;
		startJob( quarter, year);

	}
}
